package org.argus.job.server.remote.transporter.impl;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.argus.job.common.PowerSerializable;
import org.argus.job.common.enums.Protocol;
import org.argus.job.remote.framework.actor.Actor;
import org.argus.job.remote.framework.base.RemotingException;
import org.argus.job.remote.framework.base.URL;
import org.argus.job.remote.framework.transporter.Transporter;
import org.argus.job.remote.http.HttpVertxCSInitializer;
import org.argus.job.server.remote.transporter.ProtocolInfo;
import org.argus.job.server.remote.transporter.TransportService;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

import lombok.extern.slf4j.Slf4j;

/**
 * server 数据传输服务
 *
 * @author tjq
 * @since 2023/1/21
 */
@Slf4j
@Service
public class PowerTransportService implements TransportService, InitializingBean, ApplicationContextAware {

    /**
     * server 需要激活的通讯协议，暂时只支持http
     */
    @Value("${argus.transporter.active.protocols:HTTP}")
    private String activeProtocols;

    /**
     * 主要通讯协议，用于 server 与 server 之间的通讯，用户必须保证该协议可用（端口开放）！
     */
    @Value("${argus.transporter.main.protocol:HTTP}")
    private String mainProtocol;

    private static final String PROTOCOL_PORT_CONFIG = "argus.%s.port";

    private final Environment environment;

    private ProtocolInfo defaultProtocol;
    private final Map<String, ProtocolInfo> protocolName2Info = Maps.newHashMap();

    private ApplicationContext applicationContext;

    public PowerTransportService(Environment environment) {
        this.environment = environment;
    }

    @Override
    public ProtocolInfo defaultProtocol() {
        return defaultProtocol;
    }

    @Override
    public Map<String, ProtocolInfo> allProtocols() {
        return protocolName2Info;
    }

    private ProtocolInfo fetchProtocolInfo(String protocol) {
        // 兼容老版 worker 未上报 protocol 的情况
        protocol = compatibleProtocol(protocol);
        final ProtocolInfo protocolInfo = protocolName2Info.get(protocol);
        if (protocolInfo == null) {
            throw new IllegalArgumentException("can't find Transporter by protocol :" + protocol);
        }
        return protocolInfo;
    }

    @Override
    public void tell(String protocol, URL url, PowerSerializable request) {
        fetchProtocolInfo(protocol).getTransporter().tell(url, request);
    }

    @Override
    public <T> CompletionStage<T> ask(String protocol, URL url, PowerSerializable request, Class<T> clz)
        throws RemotingException {
        return fetchProtocolInfo(protocol).getTransporter().ask(url, request, clz);
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        log.info("[PowerTransportService] start to initialize whole PowerTransportService!");
        log.info("[PowerTransportService] activeProtocols: {}", activeProtocols);

        if (StringUtils.isEmpty(activeProtocols)) {
            throw new IllegalArgumentException("activeProtocols can't be empty!");
        }

        buildProtocolInfo();
        choseDefault();
        log.info("[PowerTransportService] initialize successfully!");
        log.info("[PowerTransportService] ALL_PROTOCOLS: {}", protocolName2Info);
    }

    private void buildProtocolInfo() {
        HttpVertxCSInitializer httpVertxCSInitializer = new HttpVertxCSInitializer();
        httpVertxCSInitializer.init(null);
        ProtocolInfo httpProtocolInfo =
            new ProtocolInfo(Protocol.HTTP.name(), null, httpVertxCSInitializer.buildTransporter());
        protocolName2Info.put(httpProtocolInfo.getProtocol(), httpProtocolInfo);
    }

    private String compatibleProtocol(String p) {
        if (p == null) {
            return Protocol.HTTP.name();
        }
        return p;
    }

    /**
     * HTTP 优先，否则默认取第一个协议
     */
    private void choseDefault() {

        this.defaultProtocol = this.protocolName2Info.get(mainProtocol);
        log.info("[PowerTransportService] chose [{}] as the default protocol, make sure this protocol can work!",
            mainProtocol);

        if (this.defaultProtocol == null) {
            throw new IllegalArgumentException("can't find default protocol, please check your config!");
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
